
#include "tcp.h"

#include <signal.h>

static uint16_t init_socket (p_main_flow mflow)
{
	p_socket_io sock = &(mflow->m_socket_io);
	p_socket_state sstate = &(mflow->m_flow_state.sstate);
	init_sockaddr_in_2(&(sock->local_addr), sstate->local_ip, sstate->local_port, 1);
	init_sockaddr_in_2(&(sock->remote_addr), sstate->remote_ip, sstate->remote_port, 1);
	init_ip_mreq_2(&(sock->mreq), sstate->local_ip, sstate->mreq_ip);

	sock->fd = socket(AF_INET, SOCK_STREAM, 0);
	if (sock->fd < 0) {
		printf("socket create failed\n");
		return ERR_SOCKET_CREATE;
	}
	//int old_flags = fcntl(sock->fd, F_GETFL);
	//fcntl(sock->fd, F_SETFL, old_flags | O_NONBLOCK);
	
	int optval = 1;
	optval = 10*1024*1024;
	setsockopt(sock->fd, SOL_SOCKET, SO_SNDBUF, &optval, sizeof(optval));
	setsockopt(sock->fd, SOL_SOCKET, SO_RCVBUF, &optval, sizeof(optval));
	int reuse = 1;
	setsockopt(sock->fd, SOL_SOCKET, SO_REUSEADDR, (const char*)&reuse,sizeof(int));

	struct sockaddr_in loc;
	memcpy(&loc, &sock->local_addr, sizeof(sock->local_addr));
	loc.sin_addr.s_addr=INADDR_ANY;
	if(bind(sock->fd, (const struct sockaddr *)&loc, sizeof(loc))==-1)
	{
		printf("error in binding:%s:%d\n", inet_ntoa(loc.sin_addr), ntohs(loc.sin_port));
		return ERR_SOCKET_BIND;
	}
	printf("tcp:binding:%s:%d\n", inet_ntoa(loc.sin_addr), ntohs(loc.sin_port));
	mflow->m_clean_flow = tcp_clean;
	return NO_ERR;
}

uint16_t tcp_client(p_main_flow mflow)
{
	uint16_t err;
	char buf[MAX_MESSAGE_LEN] = {0};
	int16_t data_size = 0;
	int i;
	struct tcp_info info;
	int tcp_info_len = sizeof(info);
	for (i = 0; i < mflow->m_data_size+1; i++) {
		buf[i] = i%256;
	}

	err = init_socket(mflow);
	if (err != NO_ERR) {
		return err;
	}

	printf("waiting...connect\n");
	while (connect(mflow->m_socket_io.fd, (struct sockaddr *)&(mflow->m_socket_io.remote_addr), sizeof(struct sockaddr)) < 0) {
		printf("Connect failed: %s:%d\n", inet_ntoa(mflow->m_socket_io.remote_addr.sin_addr), ntohs(mflow->m_socket_io.remote_addr.sin_port));
		if (mflow->m_bstop) break;
		sleep(1);
	}
	printf("tcp:conneck ok: %s:%d\n", inet_ntoa(mflow->m_socket_io.remote_addr.sin_addr), ntohs(mflow->m_socket_io.remote_addr.sin_port));
	while(!(mflow->m_bstop)) {
		//getsockopt(mflow->m_socket_io.fd, IPPROTO_TCP, TCP_INFO, &info, (socklen_t *)&tcp_info_len);
		//if (info.tcpi_state != TCP_CLOSE) {
		//	close(mflow->m_socket_io.fd);
		//	break;
		//}
		if (mflow->m_check_pause && mflow->m_check_pause(mflow)) {
			sleep(1);
			continue;
		}
		data_size = send(mflow->m_socket_io.fd, buf, mflow->m_data_size, 0);
		if (data_size <= 0) {
			update_input_state(mflow, 1, data_size, ERR_SEND);
		} else if (data_size < mflow->m_data_size) {
			mflow->m_flow_state.tx_pkt += 1;
			mflow->m_flow_state.tx_byte += data_size;
			mflow->m_flow_state.err_tx_pkt += 1;
			mflow->m_flow_state.err_tx_byte += mflow->m_data_size - data_size;
			//update_input_state(mflow, 1, data_size, NO_ERR);
			//update_input_state(mflow, 1, mflow->m_data_size - data_size, ERR_SEND);
		} else {
			mflow->m_flow_state.tx_pkt += 1;
			mflow->m_flow_state.tx_byte += data_size;
			//update_input_state(mflow, 1, data_size, NO_ERR);
		}
	}
	close(mflow->m_socket_io.fd);

	return NO_ERR;
}

uint16_t tcp_server(p_main_flow mflow)
{
	int clientfd = -1;
	int i;
	char buffer[MAX_MESSAGE_LEN] = {0};
	int k;
	uint16_t err;

	err = init_socket(mflow);
	if (err != NO_ERR) {
		return err;
	}

	listen(mflow->m_socket_io.fd, 10);

	struct sockaddr_in clientaddr;
	socklen_t len = sizeof(clientaddr);
#if 1
	struct tcp_info info;
	int tcp_info_len = sizeof(info);
	int flags = fcntl(mflow->m_socket_io.fd, F_GETFL, 0);
	fcntl(mflow->m_socket_io.fd, F_SETFL, flags | O_NONBLOCK);
	while(!(mflow->m_bstop)) {
		if ((clientfd = accept(mflow->m_socket_io.fd, (struct sockaddr*)&clientaddr, &len)) < 0) continue;
		printf("client: %s:%d entry", inet_ntoa(clientaddr.sin_addr), ntohs(clientaddr.sin_port));
		while(!(mflow->m_bstop)) {
			//getsockopt(clientfd, IPPROTO_TCP, TCP_INFO, &info, (socklen_t *)&tcp_info_len);
			//if (info.tcpi_state != TCP_ESTABLISHED) {
			//	printf("client: %s:%d release", inet_ntoa(clientaddr.sin_addr), ntohs(clientaddr.sin_port));
			//	close(clientfd);
			//	break;
			//}
			k = recv(clientfd, buffer, MAX_MESSAGE_LEN, 0);
			if(k <= 0) {
				break;
			}
			mflow->m_flow_state.rx_pkt += 1;
			mflow->m_flow_state.rx_byte += k;
			//update_output_state(mflow, 1, k, NO_ERR);
		}
		close(clientfd);
	}
	return NO_ERR;
#endif
	int epfd = epoll_create(1);

	struct epoll_event ev;
	ev.events = EPOLLIN;
	ev.data.fd = mflow->m_socket_io.fd;

	epoll_ctl(epfd, EPOLL_CTL_ADD, mflow->m_socket_io.fd, &ev);

	struct epoll_event events[1024] = {0};

	while(!(mflow->m_bstop))
	{
		int nready = epoll_wait(epfd, events, 1024, -1);
		if(nready < 0)
			continue;

		for(i = 0; i < nready; i++)
		{
			if((mflow->m_socket_io.fd == events[i].data.fd) && (events[i].events & EPOLLIN))
			{
				clientfd = accept(mflow->m_socket_io.fd, (struct sockaddr*)&clientaddr, &len);
				if(clientfd < 0)
					continue;

				//ev.events = EPOLLIN | EPOLLET;
				ev.events = EPOLLIN;
				ev.data.fd = clientfd;
				epoll_ctl(epfd, EPOLL_CTL_ADD, clientfd, &ev);
			} else {
				if(events[i].events & EPOLLIN)
				{
					k = recv(events[i].data.fd, buffer, MAX_MESSAGE_LEN, 0);
					if(k <= 0) {
						continue;
					}
					mflow->m_flow_state.rx_pkt += 1;
					mflow->m_flow_state.rx_byte += k;
					update_output_state(mflow, 1, k, err);

				} else if ((EPOLLERR & events[i].events) || (events[i].events & EPOLLHUP)) {
					epoll_ctl(epfd, EPOLL_CTL_DEL, events[i].data.fd, NULL);
					close(events[i].data.fd);
				}
			}
		}
	}
	close(mflow->m_socket_io.fd);
	return NO_ERR;
}

void tcp_clean(void* arg)
{
	p_main_flow mflow = (p_main_flow)(arg);
	if (!mflow) {
		printf("parameter err\n");
		return ;
	}
	close(mflow->m_socket_io.fd);
}

static int first_send=0;
uint16_t tcp_send(p_socket_io sock, char* buf, uint16_t* size)
{
	int _size = 0;
	do {
		_size = send(sock->fd, buf, *size, 0);
	} while (0);
	if (*size > _size) {
		*size = _size;
		return ERR_SEND;
	}
	*size = _size;
	return NO_ERR;
}


uint16_t tcp_recv(p_socket_io sock, char* buf, uint16_t* size)
{
	int _size = 0;
	_size = recv(sock->fd, buf, *size, 0);
	*size = _size;
	return NO_ERR;
}

